import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import scala.collection.mutable

/**
 * @author canfeit
 */
object Hbases {
  val table = new mutable.HashMap[String, HTable]()
  val conf = HBaseConfiguration.create()

  def getTable(tableName: String): HTable = {

    table.getOrElse(tableName, {
      println("----new connection ----")
      val tbl = new HTable(conf, tableName)
      table(tableName) = tbl
      tbl
    })
  }

  def selectByFilter(tablename: String, arr: List[Array[String]]) {
    val table = getTable(tablename)
    val filterList = new FilterList()
    val s1 = new Scan()
    for (s <- arr) { // 各个条件之间是“与”的关系
      val filter = new SingleColumnValueFilter(Bytes
        .toBytes(s(0)), Bytes.toBytes(s(1)), CompareOp.EQUAL, Bytes
        .toBytes(s(2)))
      filter.setFilterIfMissing(true)
      filterList.addFilter(filter)
    }
    s1.setFilter(filterList);
    val ResultScannerFilterList = table.getScanner(s1)
    var r = ResultScannerFilterList.next()

    while (r != null) {
      var data = r.getRow()
      var offset = 0;
      var length = 0;
      var datas = new String(data);
      System.out.println();
      System.out.println(datas);

      val rc = r.rawCells()
      for (cell <- rc) {
        data = cell.getFamilyArray();
        offset = cell.getFamilyOffset();
        length = cell.getFamilyLength();
        datas = new String(data, offset, length);
        System.out.print(datas + ":");

        data = cell.getQualifierArray();
        offset = cell.getQualifierOffset();
        length = cell.getQualifierLength();
        datas = new String(data, offset, length);
        System.out.print(datas + ":");

        data = cell.getValueArray();
        offset = cell.getValueOffset();
        length = cell.getValueLength();
        datas = new String(data, offset, length);
        System.out.println(datas);

      }
      r = ResultScannerFilterList.next()
    }
    table.close();
  }

  def main(args: Array[String]) {
    val tableName = "test1"
    var arr = Array("familys", "column2", "value1") :: Nil
   // arr = Array("familys", "column", "value1") :: arr
    selectByFilter(tableName, arr)
  }
}